package com.heroh.kettle.service.impl;

import com.heroh.kettle.entity.ParamDbConnection;
import com.heroh.kettle.service.IKettleService;
import com.heroh.kettle.util.KettleJobUtil;
import com.heroh.kettle.util.TransBuild;
import com.heroh.kettle.util.MetaBuild;
import lombok.extern.slf4j.Slf4j;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.logging.LogLevel;
import org.pentaho.di.job.Job;
import org.pentaho.di.job.JobMeta;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.Map;

/**
 * 包名称：com.win.dfbp.info.kettle.service.impl
 * 类名称：KettleServiceImpl
 * 类描述：kettle 服务实现类
 * 创建人：@author jianshengxiong
 * 创建时间：2020/6/1/11:08
 */
@Service
@Slf4j
public class KettleServiceImpl implements IKettleService {

    @Override
    public void runTransFile(String fileName, Map<String, String> params, Map<String, String> dbMap,List<ParamDbConnection> conns) {
        try {
            TransMeta transMeta = new MetaBuild(fileName)
                    .setConns(conns,dbMap)
                    .setTransLog()
                    .build();
            Trans trans = new TransBuild().build(transMeta);
            // 设置变量(这些变量可以在.ktl中通过${variableName}来引用)
            for (String key : params.keySet()) {
                trans.setVariable(key, params.get(key));
            }
            trans.setLogLevel(LogLevel.ROWLEVEL);
            trans.execute(null);
            trans.waitUntilFinished();
            if (trans.getErrors() > 0) {
                log.error("kettle转换异常");
            } else {
                log.info("执行成功");
            }
        } catch (KettleException e) {
        }
    }

    @Override
    public JobMeta initJobMeta(List<ParamDbConnection> conns) {
        return KettleJobUtil.initJobMeta(conns);
    }

    @Override
    public void runJob(JobMeta jobMeta) {
        // jobname 是Job脚本的路径及名称
        Job job = new Job(null, jobMeta);
        job.start();
        job.waitUntilFinished();
        if (job.getErrors() > 0) {
            log.error("kettle job异常");
//            throw WinExceptionUtil.winException(SchExceptionEnum.KETTLE_JOB_ERROR);
        } else {
            log.info("执行成功");
        }
    }

    @Override
    public void runJobFile(String fileName, Map<String, String> params, Map<String, String> dbMap, List<ParamDbConnection> conns) {
        JobMeta jobMeta = KettleJobUtil.initJobMeta(fileName,conns,dbMap);
        Job job = new Job(null, jobMeta);
        // 设置变量(这些变量可以在.ktl中通过${variableName}来引用)
        for (String key : params.keySet()) {
            job.setVariable(key, params.get(key));
        }
        job.start();
        job.waitUntilFinished();
        if (job.getErrors() > 0) {
            log.error("kettle job异常");
//            throw WinExceptionUtil.winException(SchExceptionEnum.KETTLE_JOB_ERROR);
        } else {
            log.info("执行成功");
        }
    }
}
